package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.DwsTrafficForSourcePvBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import com.atguigu.edu.realtime.common.util.FlinkSourceUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

/**
 * ClassName: DwsTrafficVcSourceArIsNewPageViewWindow
 * Description:
 *
 */
public class DwsTrafficVcSourceArIsNewPageViewWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsTrafficVcSourceArIsNewPageViewWindow().start(
                10022,
                4,
                "dws_traffic_vc_sc_ar_is_new_page_view_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaSourceDS) {
        //TODO 1.将page日志转换为jsonObj
        SingleOutputStreamOperator<JSONObject> pageJsonObjDS = kafkaSourceDS.map(JSON::parseObject);


        //TODO 2.读取用户跳出明细表，封装为流，转换结构
        KafkaSource<String> userJumpKfSource =
                FlinkSourceUtil.getKafkaSource("dwd_traffic_user_jump", "user_jump_dws_traffic_vc_sc_ar_is_new_page_view_window");

        DataStreamSource<String> userJumpDS = env.fromSource(userJumpKfSource, WatermarkStrategy.noWatermarks(), "user_jump");

        SingleOutputStreamOperator<JSONObject> userJumpJsonObj = userJumpDS.map(JSON::parseObject);

        //userJumpDS.print();


        //TODO 3.page流按照mid分组
        KeyedStream<JSONObject, String> pageMidKeyedDS = pageJsonObjDS.keyBy(j -> j.getJSONObject("common").getString("mid"));


        //TODO 4.分别将两条流数据封装为实体类对象
        //4.1 page流
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> pageBeanDS = pageMidKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, DwsTrafficForSourcePvBean>() {

                    private ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateProperties = new ValueStateDescriptor<String>("lastVisitDateState", String.class);
                        stateProperties.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDateState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, DwsTrafficForSourcePvBean>.Context ctx, Collector<DwsTrafficForSourcePvBean> out) throws Exception {
                        JSONObject common = jsonObj.getJSONObject("common");
                        JSONObject page = jsonObj.getJSONObject("page");

                        //获取当前时间
                        Long ts = jsonObj.getLong("ts");
                        String curDate = DateFormatUtil.tsToDate(ts);


                        String vc = common.getString("vc");
                        String sc = common.getString("sc");
                        String ar = common.getString("ar");
                        String isNew = common.getString("is_new");
                        Long duringTime = page.getLong("during_time");

                        //从状态获取上次访问日期
                        String lastVisitDate = lastVisitDateState.value();

                        //根据状态情况，赋值uvCount一次
                        long uvCt = 0L;
                        if (StringUtils.isEmpty(lastVisitDate) || !lastVisitDate.equals(curDate)) {
                            uvCt = 1L;
                            lastVisitDateState.update(curDate);
                        }

                        //根据last_page_id，赋值会话数
                        String lastPageId = page.getString("last_page_id");
                        long svCt = StringUtils.isEmpty(lastPageId) ? 1L : 0L;

                        DwsTrafficForSourcePvBean dwsTrafficForSourcePvBean = DwsTrafficForSourcePvBean
                                .builder()
                                .versionCode(vc)
                                .sourceId(sc)
                                .ar(ar)
                                .isNew(isNew)
                                .uvCount(uvCt)
                                .totalSessionCount(svCt)
                                .pageViewCount(1L)
                                .totalDuringTime(duringTime)
                                .jumpSessionCount(0L)
                                .ts(ts)
                                .build();
                        out.collect(dwsTrafficForSourcePvBean);
                    }

                }
        );

        //4.1 userJump流
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> userJumpBeanDS = userJumpJsonObj.map(
                new MapFunction<JSONObject, DwsTrafficForSourcePvBean>() {
                    @Override
                    public DwsTrafficForSourcePvBean map(JSONObject jsonObject) throws Exception {
                        JSONObject common = jsonObject.getJSONObject("common");

                        //获取当前时间
                        Long ts = jsonObject.getLong("ts");

                        String vc = common.getString("vc");
                        String sc = common.getString("sc");
                        String ar = common.getString("ar");
                        String isNew = common.getString("is_new");

                        return DwsTrafficForSourcePvBean
                                .builder()
                                .versionCode(vc)
                                .sourceId(sc)
                                .ar(ar)
                                .isNew(isNew)
                                .uvCount(0L)
                                .totalSessionCount(0L)
                                .pageViewCount(0L)
                                .totalDuringTime(0L)
                                .jumpSessionCount(1L)
                                .ts(ts)
                                .build();
                    }
                }
        );

        //TODO 5.合流
        DataStream<DwsTrafficForSourcePvBean> unionDS = pageBeanDS.union(userJumpBeanDS);
        //unionDS.print();

        //TODO 6.Watermark & 事件时间字段
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> withWatermarkDS = unionDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<DwsTrafficForSourcePvBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<DwsTrafficForSourcePvBean>() {
                                    @Override
                                    public long extractTimestamp(DwsTrafficForSourcePvBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

        //TODO 7.按照维度分组
        KeyedStream<DwsTrafficForSourcePvBean, Tuple4<String, String, String, String>> keyedDS = withWatermarkDS.keyBy(
                new KeySelector<DwsTrafficForSourcePvBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(DwsTrafficForSourcePvBean value) throws Exception {
                        return Tuple4.of(
                                value.getVersionCode(),
                                value.getSourceId(),
                                value.getAr(),
                                value.getIsNew()
                        );
                    }
                }
        );

        //TODO 8.开窗
        WindowedStream<DwsTrafficForSourcePvBean, Tuple4<String, String, String, String>, TimeWindow> windowDS = 
                keyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)));

        //TODO 9.聚合
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> reduceDS = windowDS.reduce(
                new ReduceFunction<DwsTrafficForSourcePvBean>() {
                    @Override
                    public DwsTrafficForSourcePvBean reduce(DwsTrafficForSourcePvBean value1, DwsTrafficForSourcePvBean value2) throws Exception {
                        //System.out.println("reduce");
                        
                        value1.setUvCount(value1.getUvCount() + value2.getUvCount());
                        value1.setTotalSessionCount(value1.getTotalSessionCount() + value2.getTotalSessionCount());
                        value1.setPageViewCount(value1.getPageViewCount() + value2.getPageViewCount());
                        value1.setTotalDuringTime(value1.getTotalDuringTime() + value2.getTotalDuringTime());
                        value1.setJumpSessionCount(value1.getJumpSessionCount() + value2.getJumpSessionCount());
                        return value1;
                    }
                },
                new ProcessWindowFunction<DwsTrafficForSourcePvBean, DwsTrafficForSourcePvBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple4<String, String, String, String> tp4,
                                        ProcessWindowFunction<DwsTrafficForSourcePvBean, DwsTrafficForSourcePvBean, Tuple4<String, String, String, String>, TimeWindow>.Context context,
                                        Iterable<DwsTrafficForSourcePvBean> elements,
                                        Collector<DwsTrafficForSourcePvBean> out) throws Exception {
                        //System.out.println("process");
                        
                        DwsTrafficForSourcePvBean bean = elements.iterator().next();

                        String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDate = DateFormatUtil.tsToDate(context.window().getStart());

                        bean.setStt(stt);
                        bean.setEdt(edt);
                        bean.setCurDate(curDate);

                        out.collect(bean);
                    }
                }
        );
        //reduceDS.print();

        //TODO 10.关联维度层sc名称
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> withSourceDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<DwsTrafficForSourcePvBean>() {
                    @Override
                    public void addDims(DwsTrafficForSourcePvBean bean, JSONObject dimJsonObj) {
                        bean.setSourceName(dimJsonObj.getString("source_site"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_source";
                    }

                    @Override
                    public String getRowKey(DwsTrafficForSourcePvBean obj) {
                        return obj.getSourceId();
                    }
                },
                60L,
                TimeUnit.SECONDS
        );

        //TODO 11.关联维度层ar名称
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> withProvinceDS = AsyncDataStream.unorderedWait(
                withSourceDS,
                new DimAsyncFunction<DwsTrafficForSourcePvBean>() {
                    @Override
                    public void addDims(DwsTrafficForSourcePvBean bean, JSONObject dimJsonObj) {
                        bean.setProvinceName(dimJsonObj.getString("name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_province";
                    }

                    @Override
                    public String getRowKey(DwsTrafficForSourcePvBean obj) {
                        return obj.getAr();
                    }
                },
                60L,
                TimeUnit.SECONDS
        );

        //withProvinceDS.print();
        
        //TODO 12.将关联结果写到Doris
        withProvinceDS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_vc_sc_ar_is_new_page_view_window"));

    }
}
